package cn.saltice.idempotence.idem.idempotent;

import cn.saltice.idempotence.idem.config.IdempotentProperties;
import cn.saltice.idempotence.idem.distributed.DistributedLock;
import cn.saltice.idempotence.idem.enums.IdempotentStorageTypeEnum;
import cn.saltice.idempotence.idem.enums.ReadWriteTypeEnum;
import cn.saltice.idempotence.idem.request.IdempotentRequest;
import cn.saltice.idempotence.idem.store.IdempotentStorage;
import cn.saltice.idempotence.idem.store.IdempotentStorageFactory;
import cn.saltice.idempotence.idem.warpper.IdempotentRunnable;
import cn.saltice.idempotence.idem.warpper.IdempotentSupplier;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @program: idempotence
 * @description: 幂等实现
 * @author: Mr.Liu
 * @create: 2020-10-03 21:31
 **/
@Slf4j
public class IdempotentImpl extends AbstractIdempotent{


    public IdempotentImpl(IdempotentStorage one, IdempotentStorage two) {
        super(one,two);
    }

    @Autowired
    private DistributedLock distributedLock;
    @Autowired
    private IdempotentProperties idempotentProperties;
    @Autowired
    private IdempotentStorageFactory idempotentStorageFactory;
    /**
     * 锁名字后缀，区分锁和幂等的key
     */
    private String lockSuffix = "_lock";
    /**
     * 幂等key对应的默认值
     */
    private String idempotentDefaultValue = "1";

    @Override
    public <T> T execute(String key, int lockExpireTime, int firstLevelExpireTime, int secondLevelExpireTime, TimeUnit timeUnit, ReadWriteTypeEnum readWriteType, IdempotentSupplier<T> execute, IdempotentSupplier<T> fail) {
        IdempotentRequest idempotentRequest = IdempotentRequest.builder().key(key)
                .lockExpireTime(lockExpireTime)
                .firstLevelExpireTime(firstLevelExpireTime)
                .secondLevelExpireTime(secondLevelExpireTime)
                .timeUnit(timeUnit)
                .readWriteType(readWriteType)
                .build();
        return execute(idempotentRequest,execute,fail);
    }

    @Override
    public <T> T execute(String key, int lockExpireTime, int firstLevelExpireTime, int secondLevelExpireTime, TimeUnit timeUnit, IdempotentSupplier<T> execute, IdempotentSupplier<T> fail) {
        return execute(key, lockExpireTime, firstLevelExpireTime, secondLevelExpireTime, timeUnit,ReadWriteTypeEnum.PARALLEL, execute, fail);
    }

    /**
     * 顺序执行
     * @param firstIdempotentStorage 一级存储
     * @param secondIdempotentStorage 二级存储
     * @param request 请求
     * @param execute 成功执行的回调函数
     * @param fail 失败执行的回调函数
     * @param <T>
     * @return
     */
    private <T> T orderExecute(IdempotentStorage firstIdempotentStorage, IdempotentStorage secondIdempotentStorage,
                               IdempotentRequest request, Object execute, Object fail){
        String firstValue = firstIdempotentStorage.getValue(request.getKey());
        // 一级存储中有数据，就不用查询二级缓存了
        if (!StringUtils.hasText(firstValue)){
            // 一级存储中没有数据，查询二级存储
            String secondValue = null;
            if (secondIdempotentStorage != null){
                secondValue = secondIdempotentStorage.getValue(request.getKey());
            }
            // 一级二级存储中都没有数据，表示可以继续执行
            if (!StringUtils.hasText(secondValue)){
                // 先执行业务逻辑，因为当前是在加锁状态的，不用担心。
                T o = getExecuteResult(execute, request);
                // 先存入一级缓存
                firstIdempotentStorage.setValue(request.getKey(), idempotentDefaultValue, request.getFirstLevelExpireTime(), request.getTimeUnit());
                if (secondIdempotentStorage != null){
                    // 再存入二级缓存
                    secondIdempotentStorage.setValue(request.getKey(), idempotentDefaultValue, request.getSecondLevelExpireTime(), request.getTimeUnit());
                }
                return o;
            }
        }
        // 不能继续往下执行
        return getExecuteResult(fail, request);
    }
    /**
     * 并行执行
     * @param firstIdempotentStorage 一级存储
     * @param secondIdempotentStorage 二级存储
     * @param request 请求
     * @param execute 成功执行的回调函数
     * @param fail 失败执行的回调函数
     * @param <T>
     * @return
     */
    private <T> T parallelExecute(IdempotentStorage firstIdempotentStorage, IdempotentStorage secondIdempotentStorage,
                                  IdempotentRequest request, Object execute, Object fail){
        List<String> results = getParallelExecuteResult(request, firstIdempotentStorage, secondIdempotentStorage);
        if (results.stream().noneMatch(StringUtils::hasText)){
            // 先执行业务逻辑，因为当前是在加锁状态的，不用担心。
            T o = getExecuteResult(execute, request);
            // 并行写入缓存
            parallelWriteResults(firstIdempotentStorage, secondIdempotentStorage, request);
            return o;
        }
        return getExecuteResult(fail,request);
    }
    /**
     * 获得执行结果 单个的执行器
     * @param execute 执行器
     * @param request 请求
     * @param <T>
     * @return
     */
    private <T> T getExecuteResult(Object execute, IdempotentRequest request){
        T executeResult = null;
        if (execute instanceof IdempotentSupplier){
            IdempotentSupplier<T> idempotentSupplier = (IdempotentSupplier<T>) execute;
            executeResult = idempotentSupplier.get(request.getKey(),null);
        }else {
            IdempotentRunnable idempotentRunnable = (IdempotentRunnable) execute;
            idempotentRunnable.run(request.getKey(),null);
        }
        return executeResult;
    }

    /**
     * 并行执行
     * @param request 请求
     * @param firstIdempotentStorage 一级缓存
     * @param secondIdempotentStorage 二级缓存
     * @return
     */
    private List<String> getParallelExecuteResult(IdempotentRequest request, IdempotentStorage firstIdempotentStorage,
                                                  IdempotentStorage secondIdempotentStorage){
        CompletableFuture<String> firstReadFuture = CompletableFuture.supplyAsync(()->{
            return firstIdempotentStorage.getValue(request.getKey());
        });
        CompletableFuture<String> secondReadFuture = CompletableFuture.supplyAsync(()->{
            if (secondIdempotentStorage != null){
                return secondIdempotentStorage.getValue(request.getKey());
            }
            return null;
        });
        CompletableFuture<Void> readCombinedFuture = CompletableFuture.allOf(firstReadFuture,secondReadFuture);
        try {
            readCombinedFuture.get();
        }catch (Exception e){
            log.error("并行读异常",e);
        }
        return Stream.of(firstReadFuture,secondReadFuture).map(CompletableFuture::join).collect(Collectors.toList());
    }

    /**
     * 并行写
     * @param firstIdempotentStorage 一级缓存
     * @param secondIdempotentStorage 二级缓存
     * @param request 请求
     */
    private void parallelWriteResults(IdempotentStorage firstIdempotentStorage, IdempotentStorage secondIdempotentStorage,
                                      IdempotentRequest request){
        CompletableFuture<String> firstWriteFuture = CompletableFuture.supplyAsync(()->{

            firstIdempotentStorage.setValue(request.getKey(), idempotentDefaultValue,
                    request.getFirstLevelExpireTime(),request.getTimeUnit());
            return null;
        });
        CompletableFuture<String> secondWriteFuture = CompletableFuture.supplyAsync(()->{
            if (secondIdempotentStorage != null){
                if (StringUtils.hasText(idempotentProperties.getSecondLevelType())){
                    secondIdempotentStorage.setValue(request.getKey(), idempotentDefaultValue,
                            request.getSecondLevelExpireTime(), request.getTimeUnit());
                }
            }
            return null;
        });
        CompletableFuture<Void> completableFuture = CompletableFuture.allOf(firstWriteFuture,secondWriteFuture);
        try {
            completableFuture.get();
        }catch (Exception e){
            log.error("并发写异常",e);
        }
    }
    @Override
    public <T> T execute(IdempotentRequest request, IdempotentSupplier<T> execute, IdempotentSupplier<T> fail) {
        return distributedLock.lock(request.getKey()+lockSuffix,request.getLockExpireTime(),
                request.getTimeUnit(),(k,ex)->{
                    IdempotentStorage secondIdempotentStorage = null;
                    if (StringUtils.hasText(idempotentProperties.getSecondLevelType())){
                        secondIdempotentStorage = idempotentStorageFactory.getIdempotentStorage(IdempotentStorageTypeEnum.valueOf(idempotentProperties.getSecondLevelType()));
                    }
                    IdempotentStorage firstIdempotentStorage = idempotentStorageFactory.getIdempotentStorage(IdempotentStorageTypeEnum.valueOf(idempotentProperties.getFirstLevelType()));

                    switch (request.getReadWriteType()){
                        case PARALLEL:
                            return parallelExecute(firstIdempotentStorage,secondIdempotentStorage,request,execute,fail);
                        case ORDER:
                            return orderExecute(firstIdempotentStorage,secondIdempotentStorage,request,execute,fail);
                        default:
                    }
                    return fail.get(request.getKey(),ex);
                },fail);
    }

    @Override
    public void execute(String key, int lockExpireTime, int firstLevelExpireTime, int secondLevelExpireTime, TimeUnit timeUnit, ReadWriteTypeEnum readWriteType, IdempotentRunnable execute, IdempotentRunnable fail) {
        IdempotentRequest idempotentRequest = IdempotentRequest.builder().key(key)
                .lockExpireTime(lockExpireTime)
                .firstLevelExpireTime(firstLevelExpireTime)
                .secondLevelExpireTime(secondLevelExpireTime)
                .timeUnit(timeUnit)
                .readWriteType(readWriteType)
                .build();
        execute(idempotentRequest, execute, fail);
    }

    @Override
    public void execute(String key, int lockExpireTime, int firstLevelExpireTime, int secondLevelExpireTime, TimeUnit timeUnit, IdempotentRunnable execute, IdempotentRunnable fail) {
        execute(key, lockExpireTime, firstLevelExpireTime, secondLevelExpireTime, timeUnit, ReadWriteTypeEnum.PARALLEL, execute, fail);
    }

    @Override
    public void execute(IdempotentRequest request, IdempotentRunnable execute, IdempotentRunnable fail) {
        distributedLock.lock(request.getKey()+lockSuffix, request.getLockExpireTime(),request.getTimeUnit(), (k,ex)->{
            IdempotentStorage secondIdempotentStorage = null;
            if (StringUtils.hasText(idempotentProperties.getSecondLevelType())){
                secondIdempotentStorage = idempotentStorageFactory.getIdempotentStorage(IdempotentStorageTypeEnum.valueOf(idempotentProperties.getSecondLevelType()));
            }
            IdempotentStorage firstIdempotentStorage = idempotentStorageFactory.getIdempotentStorage(IdempotentStorageTypeEnum.valueOf(idempotentProperties.getFirstLevelType()));

            switch (request.getReadWriteType()){
                case PARALLEL:
                    parallelExecute(firstIdempotentStorage,secondIdempotentStorage,request,execute,fail);
                    break;
                case ORDER:
                    orderExecute(firstIdempotentStorage,secondIdempotentStorage,request,execute,fail);
                    break;
            }
            fail.run(request.getKey(),ex);
        },fail);
    }
}
